package com.changba.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.changba.schedule.mapper.TaskInfoMapper;
import com.changba.schedule.mapper.TaskLogsMapper;
import com.changba.schedule.service.ScheduleService;
import com.changba.schedule.vote.SelectMaster;
import com.chongba.cache.CacheService;
import com.chongba.constant.Constant;
import com.chongba.entity.Task;
import com.chongba.exception.ScheduleSystemException;
import com.chongba.exception.TaskNotExistException;
import com.chongba.pojo.TaskInfoEntity;
import com.chongba.pojo.TaskInfoLogs;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.*;

@Service
@Slf4j
public class ScheduleServiceImpl implements ScheduleService {

    @Autowired
    private TaskInfoMapper taskInfoMapper;

    @Autowired
    private TaskLogsMapper taskLogsMapper;

    @Autowired
    private CacheService cacheService;

    @Autowired
    @Qualifier("visibleThreadPool")
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    private Long futureTime;

    @Autowired
    private ThreadPoolTaskScheduler scheduler;

    //数据参数化，从配置文件中读取
    @Value("${chongba.delayTime}")
    private int delayTime;

    @Autowired
    private SelectMaster selectMaster;

    /**
     * 同步数据库和缓存数据
     * <p>
     * 优化①：从数据库中查询出所有数据，按照任务的执行的时间，分配到topic_或者future_这两个key中
     * 优化②：按照任务类型和优先级，先从数据库中分组查询数据，再将数据添加缓存的操作放进线程中执行
     * 优化③：任务在executeTime>5min以后执行的，依然保存在数据库，通过定时刷新方式，将数据库中executeTime<5min需要执行的刷新到future_中
     * 改造：该模块在集成springcloud后已经可以多节点部署，意味着每个节点程序启动后，该方法都会被执行
     * 因此只能是主节点才能完成该方法，其他节点只能从主节点进行数据备份
     */

    //由于@PostConstruct在程序启动的时候只会执行一次，需要通过定时方式，每隔一段时间执行一次同步数据的操作
    @PostConstruct
    public void SyncData() {
        //选主
        selectMaster.selectMaster(Constant.SCHEDULEPATH);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        scheduler.scheduleAtFixedRate(() -> {
            //判断节点是否已经从zookeeper目录中注册
            if (selectMaster.checkMaster(Constant.SCHEDULEPATH)) {
                reloadFromDb();
                log.info("主节点在同步数据");
            } else {
                log.info("从节点备用");
            }
        }, TimeUnit.MINUTES.toMillis(1));

        /* //优化①
        List<TaskInfoEntity> taskInfoEntities = taskInfoMapper.selectList(null);
        for (TaskInfoEntity taskInfoEntity : taskInfoEntities) {
            if (taskInfoEntity != null && !StringUtils.isEmpty(taskInfoEntity)) {
                task.setExecuteTime(taskInfoEntity.getExecuteTime().getTime());
                BeanUtils.copyProperties(taskInfoEntity, task);
                addTaskToCache(task);
            }
        }*/
    }

    private void reloadFromDb() {
        System.out.println("缓存：开始同步数据库数据......");
//        System.out.println(delayTime);
        //清空缓存
        clearCache();
        //同步数据
        //优化②
        //根据任务优先级和任务类型，进行分组查询
        QueryWrapper<TaskInfoEntity> queryWrapper = new QueryWrapper<>();
        queryWrapper.select("task_type", "priority");
        queryWrapper.groupBy("task_type", "priority");
        List<Map<String, Object>> taskGroup = taskInfoMapper.selectMaps(queryWrapper);

        //统计使用线程后，程序恢复数据的时间
        long start = System.currentTimeMillis();
        CountDownLatch countDownLatch = new CountDownLatch(taskGroup.size());

        //定义任务未来执行时间
        Calendar calendar = Calendar.getInstance();
        //方便测试，设置1分钟
        calendar.add(Calendar.MINUTE, delayTime);
        futureTime = calendar.getTimeInMillis();
        //将futureTime存入redis，实现朱从节点数据共享
        cacheService.set(Constant.FUTURETIME, futureTime + "");

        for (Map<String, Object> taskInfo : taskGroup) {
            System.out.println("分组查询结果：" + taskInfo.toString());
            threadPoolTaskExecutor.execute(() -> {
                //将每一组数据添加进缓存的操作，通过线程来执行
                //根据分组结果，查询各分组内所有数据
                String task_type = String.valueOf(taskInfo.get("task_type"));
                String priority = String.valueOf(taskInfo.get("priority"));
                //优化③
                List<TaskInfoEntity> taskInfoEntities = taskInfoMapper.selectFuture(Integer.parseInt(task_type), Integer.parseInt(priority), futureTime);
                //List<TaskInfoEntity> taskInfoEntities = taskInfoMapper.selectAll(Integer.parseInt(task_type), Integer.parseInt(priority));
                for (TaskInfoEntity taskInfoEntity : taskInfoEntities) {
                    //数据添加进缓存
                    System.out.println("分组后的查询结果:" + taskInfoEntity.toString());
                    Task task = new Task();
                    task.setExecuteTime(taskInfoEntity.getExecuteTime().getTime());
                    BeanUtils.copyProperties(taskInfoEntity, task);
                    addTaskToCache(task);
                }
                countDownLatch.countDown();
                log.info("线程{},当前线程数{},每组恢复数据耗时{}", Thread.currentThread().getName(), countDownLatch.getCount(), System.currentTimeMillis() - start);
            });
        }

        try {
            countDownLatch.await(5, TimeUnit.MINUTES);
            log.info("数据恢复完成，总耗时:" + (System.currentTimeMillis() - start) + "毫秒");
        } catch (InterruptedException e) {
            log.error("数据恢复失败,失败原因{}", e.getMessage());
        }
    }


    /**
     * 清空缓存
     * <p>
     * 优化：当前redis有两个key，匹配key的查询方法有两种，key *(数据量大，线程会阻塞) , scan *
     */
    private void clearCache() {
        Set<String> topic = cacheService.scan(Constant.TOPIC + "*");
        cacheService.delete(topic);
        Set<String> future = cacheService.scan(Constant.FUTURE + "*");
        cacheService.delete(future);
//        cacheService.delete(Constant.DBCACHE);


    }

    /**
     * 添加任务
     * 优化①：可以将提交任务的操作放到线程中，继续优化程序性能
     * 优化②：通过线程方式添加任务
     *
     * @param task 任务对象
     * @return 任务id
     * @throws ScheduleSystemException 任务系统异常
     */
    @Override
    @Transactional
    public Long addTask(Task task) throws ScheduleSystemException {
        //优化②
        Future<Long> submit = threadPoolTaskExecutor.submit(() -> {
            //添加任务进数据库成功
            boolean result = addTaskToDb(task);
            if (result) {
                //将任务添加进缓存
                addTaskToCache(task);
            }
            return task.getTaskId();
        });
        Long taskId;
        try {
            taskId = submit.get(5, TimeUnit.SECONDS);
        } catch (Exception e) {
            log.error("添加任务失败");
            throw new ScheduleSystemException(e.getMessage());
        }
        return taskId;
    }

    /**
     * 取消任务
     *
     * @param taskId 任务id
     * @return 是否取消成功
     * @throws TaskNotExistException 任务不存在异常
     */
    @Override
    @Transactional
    public boolean cancelTask(Long taskId) throws TaskNotExistException {
        boolean result = false;
        Task task = updateTaskFromDB(taskId, Constant.CANCELLED);
        if (task != null) {
            deleteTaskFromCache(task);
            result = true;
        }
        return result;
    }

    /**
     * 从缓存中查询任务
     * <p>
     * 优化：任务已按照执行时间，分别存进topic_和future_两个key下，任务数量应当从这两个key中查询
     *
     * @param taskType 任务类型
     * @param priority 任务优先级
     * @return 返回任务对象，用于消费
     * @throws TaskNotExistException 缓存中查不到数据，则抛出异常
     */
    @Override
    public long size(int taskType, int priority) {
        String key = taskType + "_" + priority;
        long topicLength = cacheService.lRange(Constant.TOPIC + key, 0, -1).size();
        Set<String> futureLength = cacheService.zRangeAll(Constant.FUTURE + key);
//        Set<String> rangeAll = cacheService.zRangeAll(Constant.DBCACHE);
        return topicLength + futureLength.size();
    }

    /**
     * 从缓存中查出数据后，进行消费
     * <p>
     * 优化①：任务只从redis的List集合拉取
     * 优化②：将从redis中拉取任务的操作放到线程中执行
     *
     * @param taskType 任务类型
     * @param priority 任务优先级
     * @return 返回任务对象
     * @throws TaskNotExistException 任务不存在则抛出异常
     */
    @Override
    @Transactional
    public Task poll(int taskType, int priority) throws TaskNotExistException {
        //优化②
        Future<Task> submit = threadPoolTaskExecutor.submit(() -> {
            Task task = null;
            String key = taskType + "_" + priority;
            String taskJson = cacheService.lRightPop(Constant.TOPIC + key);
            if (taskJson != null && !StringUtils.isEmpty(taskJson)) {
                task = JSON.parseObject(taskJson, Task.class);
                updateTaskFromDB(task.getTaskId(), Constant.EXECUTED);
            }
            return task;
        });
        Task task;
        try {
            task = submit.get(5, TimeUnit.SECONDS);
        } catch (Exception e) {
            log.error("消费任务失败");
            throw new TaskNotExistException(e.getMessage());
        }
        return task;
        /*优化①
        Task task = null;
        String key = taskType + "_" + priority;
        String taskJson = cacheService.lRightPop(Constant.TOPIC + key);
        if (taskJson != null && !StringUtils.isEmpty(taskJson)) {
            task = JSON.parseObject(taskJson, Task.class);
            updateTaskFromDB(task.getTaskId(), Constant.EXECUTED);
        }
        return task;*/

        //将马上要执行的任务查询出来
        /*Task task = null;
        try {
            Set<String> taskJsonSet = cacheService.zRangeByScore(Constant.DBCACHE, 0, System.currentTimeMillis());
            if (taskJsonSet != null && !taskJsonSet.isEmpty()) {
                //遍历集合
                String taskJson = taskJsonSet.iterator().next();
                if (taskJson != null && !StringUtils.isEmpty(taskJson)) {
                    //从缓存中删除将要执行的任务
                    cacheService.zRemove(Constant.DBCACHE, taskJson);
                    //从数据库中删除任务，修改任务日志信息
                    task = JSON.parseObject(taskJson, Task.class);
                    updateTaskFromDB(task.getTaskId(), Constant.EXECUTED);
                }
            }
        } catch (TaskNotExistException e) {
            log.error("消费任务失败");
            throw new TaskNotExistException(e.getMessage());
        }
        return task;*/
    }

    /**
     * 将future_下要执行的任务数据刷新到topic_中
     * 步骤：获取所有含有future_的key
     * 根据这个key查询出所有将要执行的future数据
     * 根据这个future_ key 修改成 topic_ key
     * 将future数据添加进topic_ key 中
     * 删除future_下数据
     * <p>
     * 改造：当前模块已经集成springcloud可以多节点部署，也就意味着该方法在程序启动的时候
     * 多个节点都会执行该方法，可能造成数据冲突，需要对外提供接口，由外部统一调度使用
     */
    // @Scheduled(cron = "*/1 * * * * ?")
    public void refresh() {
        //优化②
        System.out.println(System.currentTimeMillis() / 1000 + "开始刷新缓存数据......");
        threadPoolTaskExecutor.execute(() -> {
            Set<String> futureKeys = cacheService.scan(Constant.FUTURE + "*");
            for (String futureKey : futureKeys) {// future_*_*
                String topicKey = Constant.TOPIC + futureKey.split(Constant.FUTURE)[1]; // topic_*_*
                Set<String> futureTasks = cacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());
                for (String futureTask : futureTasks) {
                    if (!futureTask.isEmpty()) {
                    /*cacheService.lLeftPush(topicKey,futureTask);
                    cacheService.zRemove(futureKey, futureTask);*/
                        /*
                         * 优化①：每一次操作redis都会与redis创建一次连接，由于redis是单线程，只有等当前连接任务执行完毕，才会继续执行其他操作
                         *      任务量巨大，程序会有一定延迟
                         *      redis提供了管道技术-->将多次redis操作，放在一次连接中，将所有请求处理完毕后，一次返回结果
                         *      使用管道技术能够再一次提升redis性能
                         *
                         * */
                        cacheService.refreshWithPipeLine(topicKey, futureKey, futureTasks);
                        System.out.println("线程:" + Thread.currentThread().getName() + "成功的将" + futureKey + "的任务，id为：" + JSON.parseObject(futureTask, Task.class).getTaskId() + "定时刷新到" + topicKey);
                    }
                }
            }
        });
    }

    /**
     * 封装添加任务进缓存的方法
     * <p>
     * 为什么优化：Zset本身存在缺陷，从查询效率来看，相对于List较低
     * 优化思路：将马上要执行的任务，也就是task.executeTime<=System.currentTimeMillis的任务要存进List
     * task.executeTime>System.currentTimeMillis的任务存进Zset
     * 任务消费只从List消费
     *
     * @param task 任务对象
     */
    private void addTaskToCache(Task task) {

        String key = task.getTaskType() + "_" + task.getPriority();
        System.out.println("key的后缀:" + key);
        //从redis中获取预加载时间
        futureTime = getFutureTime();
        //优化 ①
        if (task.getExecuteTime() <= System.currentTimeMillis()) {
            cacheService.lLeftPush(Constant.TOPIC + key, JSON.toJSONString(task));
        } else if (task.getExecuteTime() <= futureTime) {
            System.out.println("futureT:" + futureTime);
            cacheService.zAdd(Constant.FUTURE + key, JSON.toJSONString(task), task.getExecuteTime());
            System.out.println("成功将数据库任务id：" + task.getTaskId() + "刷新到缓存：" + Constant.FUTURE + key);
        }

//        cacheService.zAdd(Constant.DBCACHE, JSON.toJSONString(task), task.getExecuteTime());
    }

    private Long getFutureTime() {
        String futureTime = cacheService.get(Constant.FUTURETIME);
        if (futureTime != null && !StringUtils.isEmpty(futureTime)) {
            log.info("从缓存中获取预加载时间{}", futureTime);
            return Long.parseLong(futureTime);
            //redis没有这个预加载时间，做数据补偿
        } else {
            Calendar calendar = Calendar.getInstance();
            calendar.add(Calendar.MINUTE, delayTime);
            this.futureTime = calendar.getTimeInMillis();
            cacheService.set(Constant.FUTURETIME, futureTime);
            log.info("缓存中没有预加载时间，做数据补偿{}",futureTime);
            return this.futureTime;
        }
    }

    /**
     * 封装添加任务进数据库的方法
     *
     * @param task 任务对象
     * @return 是否添加成功
     */
    private boolean addTaskToDb(Task task) {
        try {
            //将Task转换为TaskInfoEntity对象
            TaskInfoEntity taskInfoEntity = new TaskInfoEntity();
            BeanUtils.copyProperties(task, taskInfoEntity);
            taskInfoEntity.setExecuteTime(new Date(task.getExecuteTime()));

            //将TaskInfoEntity信息添加进数据库
            taskInfoMapper.insert(taskInfoEntity);

            //设置任务id
            task.setTaskId(taskInfoEntity.getTaskId());

            //封装任务日志信息
            TaskInfoLogs taskInfoLogs = new TaskInfoLogs();
            BeanUtils.copyProperties(taskInfoEntity, taskInfoLogs);
            taskInfoLogs.setTaskId(taskInfoEntity.getTaskId());
            taskInfoLogs.setVersion(1);
            taskInfoLogs.setStatus(Constant.SCHEDULED);

            //将TaskInfoLog信息添加进数据库
            taskLogsMapper.insert(taskInfoLogs);

            return true;

        } catch (Exception e) {
            log.error("添加任务失败,失败任务id={}", task.getTaskId());
            throw new ScheduleSystemException(e.getMessage());
        }
    }

    /**
     * 从缓存中删除任务信息
     * <p>
     * 优化：当前任务已按照执行时间，存放在topic_和future_这两个key下，取消任务应当在这两个key中取消
     *
     * @param task 任务对象
     */
    private void deleteTaskFromCache(Task task) {
        String key = task.getTaskType() + "_" + task.getPriority();
        //根据任务执行时间，判断从哪个key中删除任务
        if (task.getExecuteTime() <= System.currentTimeMillis()) {
            cacheService.delete(Constant.TOPIC + key);
        } else {
            cacheService.delete(Constant.FUTURE + key);
        }

//        cacheService.zRemove(Constant.DBCACHE, JSON.toJSONString(task));

    }

    /**
     * 从任务表中删除任务
     *
     * @param taskId 任务id
     * @return 删除任务成功后返回任务对象，用于从缓存中删除任务信息
     */
    private Task updateTaskFromDB(Long taskId, int status) {
        Task task = null;
        try {
            //删除任务表信息
            taskInfoMapper.deleteById(taskId);
            //修改任务日志表任务状态
            TaskInfoLogs taskInfoLogs = taskLogsMapper.selectById(taskId);
            if (taskInfoLogs != null) {
                taskInfoLogs.setStatus(status);
                taskLogsMapper.updateById(taskInfoLogs);
                task = new Task();
                BeanUtils.copyProperties(taskInfoLogs, task);
                task.setExecuteTime(taskInfoLogs.getExecuteTime().getTime());
            }
        } catch (Exception e) {
            log.error("任务删除失败，失败任务id={}", taskId);
            throw new TaskNotExistException(e.getMessage());
        }
        return task;
    }
}
